package juc.thread.threadPool;

import java.util.ArrayDeque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/*
* 自定义线程池
* */
public class Test1 {
    public static void main(String[] args) {
        ThreadPool threadPool = new ThreadPool(3, 0, 3, TimeUnit.SECONDS,
                ((taskQueue, task) -> {
                    //死等
//                    taskQueue.put(task);
                    //超时等待
//                    taskQueue.offer(task,3,TimeUnit.SECONDS);
                    //直接拒绝
//                    System.out.println("直接拒绝啥也不做...");
                    //抛出异常
//                    System.out.println("执行拒绝策略...");
//                    throw new RuntimeException("任务队列已满...");
                    //由用户线程自己执行
                    task.run();

                }));
        for (int i = 0; i < 15; i++) {//任务队列满的时候 主线程一直阻塞 不友好
             int t=i;
             threadPool.excutor(() -> {
                 System.out.println(t+"任务执行...");
                 try {
                     Thread.sleep(100000);
                 } catch (InterruptedException e) {
                     e.printStackTrace();
                 }
             });
        }
    }
}
@FunctionalInterface
interface RejectPolicy<T>{
    void reject(BlockQueue<T>taskQueue,T task);
}
class ThreadPool{
    private BlockQueue<Runnable>tasksQueue;
    private HashSet<Worker>workSet;
    private final int CORESIZE;
    private int blockQueueCapacity=10;
    private long timeout;
    private TimeUnit timeUnit;
    private RejectPolicy<Runnable> rejectPolicy;

    public ThreadPool(int coreSize, int blockQueueCapacity,long timeout,TimeUnit timeUnit,RejectPolicy<Runnable> rejectPolicy) {
        this.CORESIZE = coreSize;
        if(blockQueueCapacity>this.blockQueueCapacity)
        this.blockQueueCapacity = blockQueueCapacity;
        this.timeout=timeout;
        this.timeUnit=timeUnit;
        workSet=new HashSet<>(coreSize);
        tasksQueue=new BlockQueue<>(this.blockQueueCapacity);
        this.rejectPolicy=rejectPolicy;
    }
    public void excutor(Runnable task){
        synchronized (workSet) {
            if(workSet.size()<CORESIZE){
                Worker worker = new Worker(task);
                System.out.println(task+"开始执行...");
                workSet.add(worker);
                worker.start();
            }else {
                /* 当任务繁忙时 队列就会被塞满 主线程就会阻塞 而这并不友好 而这时应该让用户决定该如何处理 策略模式*/
                tasksQueue.tryPut(task,rejectPolicy);
            }
        }
    }

    class Worker extends Thread{
        private Runnable task;
        public Worker(Runnable task){
             this.task=task;
        }
        @Override
        public void run() {
            while (task!=null||(task=tasksQueue.poll(timeout,timeUnit))!=null){
                try {
                    task.run();
                    System.out.println(task+"执行完毕...");
                } catch (Exception e) {
                    e.printStackTrace();
                }finally {
                    task=null;
                }
            }
            synchronized (workSet){
                System.out.println("空闲线程销毁"+this.getName());
                workSet.remove(this);
            }
        }
    }
}
final class BlockQueue<T>{
    private ArrayDeque<T> blockingQueue=new ArrayDeque<>();
    private final ReentrantLock lock=new ReentrantLock();
    private final Condition emptyWaitSet=lock.newCondition();
    private final Condition fullWaitSet=lock.newCondition();
    private final int capacity;
    public BlockQueue(int capacity) {
        this.capacity = capacity;
    }
    public int getQueueSize(){
          return blockingQueue.size();
    }
    public boolean offer(T t,long timeout,TimeUnit timeUnit){
        lock.lock();
        try {
            timeout=timeUnit.toNanos(timeout);
            while (blockingQueue.size()==capacity){
                try {
                    if(timeout<=0){
                        System.out.println(t+"不等了...");
                        return false;
                    }
                    System.out.println(t+"等待加入队列...任务队列满");
                    timeout = fullWaitSet.awaitNanos(timeout);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            System.out.println(t+"加入队列...");
            blockingQueue.addLast(t);
            //随机唤醒一个需要获取任务的线程
            emptyWaitSet.signal();
            return true;
        }finally {
            lock.unlock();
        }
    }
    public void put(T t){
          lock.lock();
          try {
              while (blockingQueue.size()==capacity){
                  try {
                      System.out.println(t+"等待加入队列...任务队列满");
                      fullWaitSet.await();
                  } catch (InterruptedException e) {
                      e.printStackTrace();
                  }
              }
              System.out.println(t+"加入队列...");
              blockingQueue.addLast(t);
              //随机唤醒一个需要获取任务的线程
              emptyWaitSet.signal();
          }finally {
              lock.unlock();
          }
    }
    public T poll(long timeout, TimeUnit timeUnit){
        lock.lock();
        try {
            timeout=timeUnit.toNanos(timeout);
            while (blockingQueue.isEmpty()) {
                try {
                    if(timeout<=0)return null;
                    timeout = emptyWaitSet.awaitNanos(timeout);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = blockingQueue.removeFirst();
            //随机唤醒一个不能提交任务的线程
            fullWaitSet.signal();
            return t;
        }finally {
            lock.unlock();
        }
    }
    public T take(){
        lock.lock();
        try {
            while (blockingQueue.isEmpty()) {
                try {
                    emptyWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = blockingQueue.removeFirst();
            //随机唤醒一个不能提交任务的线程
            fullWaitSet.signal();
            return t;
        }finally {
            lock.unlock();
        }
    }

    public void tryPut(T task, RejectPolicy<T> rejectPolicy){
        lock.lock();
        try {
            if(blockingQueue.size()==capacity){
                //拒绝策略 策略模式
                rejectPolicy.reject(this,task);
            }else {
                System.out.println(task+"加入任务队列...");
                blockingQueue.addLast(task);
                emptyWaitSet.signal();
            }
        } finally {
          lock.unlock();
        }
    }
}
